package concurrent;

import java.util.HashSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * Diese Klasse teilt Maschen von einer Masche ausgehend, indem sie die Vorgaenger ihrer
 * eigenen Knoten markiert und sich deren Maschen merkt. Fuer diese Maschen wird dann ueber
 * prueft ob alle Knoten darin markiert sind, dann tun wir nichts, oder nur ein Teil markiert
 * ist, dann Teilen wir diese Masche indem wir die markierten Knoten in der Masche in eine
 * neue Masche verschieben und anschliesend einen Arbeitsauftrag fuer beide losschicken
 * @author Lukas Subel, Patrick Bender
 */
public class Splitter implements Runnable {

	//Eigene Variablen
	private Masche masche;
	
	//Gemeinsam genutzte Variablen
	private ExecutorService executor;
	private HashSet<Masche> listeAllerMaschen;
	private CountUpAndDownLatch cutLatch;
	private ExecutorService hilfsExecutor;
	
	/**
	 * Konstruktor; einzig act ist individuell, Rest gemeinsam
	 * @param masche Die zu bearbeitende Masche
	 * @param executor wird benutzt um neue Splitter-Threads zu starten
	 * @param hilfsExecutor wird benutzt um neue Kontroll-Threads zu starten
	 * @param listeAllerMaschen enthaelt die Liste aller Maschen
	 * @param cutLatch zaehlt aktuell Laufende Splitter-Threads
	 */
	public Splitter(Masche masche, ExecutorService executor, ExecutorService hilfsExecutor, HashSet<Masche> listeAllerMaschen, CountUpAndDownLatch cutLatch) {
		this.masche = masche;
		this.executor = executor;
		this.hilfsExecutor = hilfsExecutor;
		this.listeAllerMaschen = listeAllerMaschen;
		this.cutLatch = cutLatch;
	}

	@Override
	public void run() {
		try {
			//Initialisierung von variablen
			
			//Gibt an ob wir alle Locks haben koennen.
			boolean getAllLocks = false;
			//Speichert von dieser Masche markierte Knoten
			HashSet<Node> marked = null;
			//Enhaelt die haelfte der Maschen mit markierten Knoten
			HashSet<Masche> maschen = null;
			//Enhaelt die andere haelfte der Maschen mit markierten Knoten
			HashSet<Masche> maschen2 = null;
			//Gibt an ob der Kontroll-Thread (wenn einer gestartet wurde) terminiert ist
			CountDownLatch unterLatch = null;
				
			//Dieser grosse Block ueberprueft ob wir alle Locks haben koennen die wir brauchen.
			//Wenn ja nehmen wir uns alle Locks die wir brauchen, wenn nicht dann geben wir 
			//nehmen wir keine locks und legen uns schlafen
			synchronized (listeAllerMaschen) {
				while (!getAllLocks) {
					getAllLocks = true;
					
					//Zunaechst pruefen wir ob wir unser eigenes Lock bekommen koennen um ungestoert
					//die markierten Knoten und deren Maschen zu bestimmen.
					if (masche.isLocked())
						getAllLocks = false;
					else {
						//Initialisierung von Variablen
						marked = new HashSet<Node>();
						maschen = new HashSet<Masche>();
						maschen2 = new HashSet<Masche>();
						//i wird genutzt um die Maschen mit markierten Knoten gleichmaesig auf
						//maschen und maschen2 zu verteilen
						int i = 0;
						//Wir iterieren ueber alle unsere Knoten
						for(Node node : masche.getNodes()) {
							//Und dann noch ueber deren Vorgaengerliste
							for(Node act : node.getPred()) { 
								//Um zu ueberpruefen ob wir alle Locks der Vorgaengerknoten bekommen
								//koennen. Lock races verhindern wir indem man immer erst das implizite
								//lock der listeAllerMaschen haben muss
								if (act.isMascheLocked() && act.getMasche() != masche) {
									//Koennen wir das lock mindestens einer Masche nicht bekommen
									//brechen wir ab
									getAllLocks = false;
									break;
								}
								else if ( act.getMasche().getSize() > 1) {
									//Ansonsten teilen wir die Maschen auf maschen und maschen2 gleichmaessig auf
									switch(i%2) {
									case 0: 
										if (!maschen2.contains(act.getMasche())) {
											maschen.add( act.getMasche() ); 
											i++;
										}
										break;
									case 1: 
										if (!maschen.contains(act.getMasche())) {
											maschen2.add( act.getMasche() ); 
											i++;
										}
									break;
									}
									//Und speichern uns die markierten knoten in marked
									marked.add(act);
								}
							}
							//Wir koennen abbrechen wenn wir mindestens ein lock nicht erhalten koennen
							if (!getAllLocks) break;
						}
					}
					
					//Sollten wir aber alle locks bekommen koennen
					if (getAllLocks) {
						//Starten wir fuer die eine Haelfte der einen Extra thread sofern maschen2 mehr als
						//eine Masche enthaelt
						if ( maschen2.size() > 1) {
							//Wir brauchen einen Locklatch um sicherzugehn das wir das implizite Lock von
							//der listeAllerMaschen erst zurueckgeben wenn wir wirklich alle locks der Maschen haben
							CountDownLatch lockLatch = new CountDownLatch(1);
							//Der unterlatch gibt an ob der hilfsThread terminiert ist
							unterLatch = new CountDownLatch(1);
							hilfsExecutor.execute( new Kontroll(maschen2, marked, listeAllerMaschen, cutLatch, unterLatch, executor, hilfsExecutor, lockLatch) );
							//Hier warten wir das der hilfsThread die Locks genommen hat
							lockLatch.await();
						}
						else {
							//Sollte machen2 weniger als 2 maschen enthalten ueberlassen wir die Arbeit diesem Thread
							//indem wir maschen2 zu maschen hinzufuegen
							maschen.addAll(maschen2);
							//In diesem fall brauchen wir auch keinen unterlatch
							unterLatch = new CountDownLatch(0);
						}
	
						//Dann nimmt dieser Thread die Locks seiner Maschen an sich
						for (Masche actMasche : maschen) {
							actMasche.lock();
						}
					}
					else
						//Haben wir nicht alle Locks warten wir einfach bis wir benachrichtigt werden
						listeAllerMaschen.wait();
				}
			//Rueckgabe des impliziten Lock von listeAllerMaschen
			}
			
			//Hier bearbeiten wir unsere Maschen
			bearbeiteMaschen(maschen, marked, executor, hilfsExecutor, listeAllerMaschen, cutLatch);
		
			//Hier warten wir auf terminierung des (eventuell gestarteten) hilfsThreads
			unterLatch.await();
			//Dannach sind wir fertig und zaehlen den cutLatch um eins runter und terminieren dann
			cutLatch.decrease();
		}
		catch (Exception e) {
			// Im Fehlerfall den Fehler ausgeben und Terminieren
			e.printStackTrace();
			System.exit(-1);
		}
	}

	public static void bearbeiteMaschen(HashSet<Masche> maschen, HashSet<Node> marked, ExecutorService executor, ExecutorService hilfsExecutor, HashSet<Masche> listeAllerMaschen, CountUpAndDownLatch cutLatch) {
		//Wir bearbeiten all unsere Maschen in dieser Schleife
		for (Masche actMasche : maschen) {
			//Wenn nicht alle Knoten der Maschen in marked zu finden sind muessen wir diese teilen
			if (!marked.containsAll(actMasche.getNodes())) { 
				//Dazu erstellen wir eine neue Masche
				Masche temp = new Masche();
				//Locken diese damit keiner mit der bearbeitung dieser Masche anfaengt solange wir 
				//noch Knoten umkopieren
				temp.lock();
				//Wir iterieren ueber alle markierten Knoten
				for(Node act : marked)
				{
					//Und entfernen die markierten Knoten aus der Masche und fuegen sie in die temp Masche ein
					if (actMasche.remove(act)) { 
						temp.add(act);
					}
				}
				//Sind wir damit fertig koennen wir das Lock der temp Masche zurueckgeben
				temp.unlock();
				//Und anschliesend die neue Masche in die Liste der Maschen hinzufuegen
				synchronized (listeAllerMaschen) {
					listeAllerMaschen.add(temp);
				}
				//Anschliesend starten wir noch zwei arbeitsauftraege. Einen fuer die geteilte Masche
				//und einen fuer die neue Masche
				cutLatch.increase();
				executor.execute( new Splitter(temp,executor,hilfsExecutor,listeAllerMaschen,cutLatch) );
				cutLatch.increase();
				executor.execute( new Splitter(actMasche,executor,hilfsExecutor,listeAllerMaschen,cutLatch) );
			}
			//Anschliesend koennen wir das Lock der gerade bearbeiteten Masche zurueckgeben
			actMasche.unlock();
			//Und die andern Splitter-Threads aufwecken
			synchronized (listeAllerMaschen) {
				listeAllerMaschen.notifyAll();
			}
		}
	}

}
